package com.xctech.yace.thread;

import com.xctech.yace.pojo.Message;
import com.xctech.yace.single.DataMap;
import com.xctech.yace.util.TimeUtil;
import lombok.*;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

@Data
@NoArgsConstructor
public class ProducerThread implements Runnable{

    private KafkaProducer<String,String> producer;

    private ProducerRecord<String,String> record;

    public ProducerThread(KafkaProducer<String, String> producer, ProducerRecord<String, String> record) {
        this.producer = producer;
        this.record = record;
    }

    @Override
    public void run() {

        String threadName = Thread.currentThread()+"|"+
                Thread.currentThread().getId()+"|"+Thread.currentThread().getName();

        producer.send(record,(metadata,e)->{
            if(null != e){
                e.printStackTrace();
            }
            if(null != metadata){
                System.out.println(threadName+
                        "-消息发送成功 ： " + String.format("offset: %s, partition:%s, topic:%s timestamp:%s",
                        metadata.offset(), metadata.partition(), metadata.topic(), metadata.timestamp()));
            }
        });
    }
}
